package bupt.xixi.netty;

import bupt.xixi.RpcClient;
import bupt.xixi.codec.CommonDecoder;
import bupt.xixi.codec.CommonEncoder;
import bupt.xixi.entity.RpcRequest;
import bupt.xixi.entity.RpcResponse;
import bupt.xixi.enumeration.Codec;
import bupt.xixi.registry.IServiceDiscovery;
import bupt.xixi.registry.impl.ZkServiceDiscovery;
import bupt.xixi.serializer.KryoSerializerI;
import bupt.xixi.serializer.SerializerFactory;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;

@Component
public class NettyClient implements RpcClient {

    private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);

    private final Integer READ_IDLE = 5 * Codec.IDLE;
    private final Integer WRITE_IDLE = 0;
    private final Integer READ_WRITE_IDLE = Codec.IDLE;

    private final Bootstrap bootstrap;
    private final IServiceDiscovery serviceDiscovery = new ZkServiceDiscovery();


    // 配置好Netty客户端链接
    {
        EventLoopGroup group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                 .channel(NioSocketChannel.class)
//                .option(ChannelOption.SO_KEEPALIVE, true) // tcp的心跳连接, 不好用, 不用它~
                 .handler(new ChannelInitializer<SocketChannel>() {
                     @Override
                     protected void initChannel(SocketChannel socketChannel) throws Exception {
                         ChannelPipeline pipeline = socketChannel.pipeline();
                         // head -> out1 -> out2 -> in1 -> in2 -> tail
                         // Inbound的执行顺序为  read -> in1 -> in2 -> write
                         // 解码器： RpcResponse -> byteBuf
                         pipeline.addLast(new IdleStateHandler(READ_IDLE, WRITE_IDLE, READ_WRITE_IDLE));
                         pipeline.addLast(new CommonDecoder());  // 是inBound
                         // 编码器：byteBuf -> RpcResponse
                         pipeline.addLast(new CommonEncoder(Codec.SerializeType.KYRO));  // 是outBound
                         pipeline.addLast(new NettyClientHandler());
                     }
                 });

    }

    /**
     * send的时候自动开启一个Netty连接
     */
    @Override
    public Object sendRequest(RpcRequest rpcRequest) {
        try {
            InetSocketAddress inetSocketAddress = serviceDiscovery.lookupService(rpcRequest.getInterfaceName());
            String host = inetSocketAddress.getHostString();
            int port = inetSocketAddress.getPort();
            ChannelFuture future = bootstrap.connect(host, port).sync();
            logger.info("客户端链接到服务器 {}:{}", host, port);
            Channel channel = future.channel();  // 得到真正用于传输的通道
            if (channel != null) {
                // 发送函数并设置回调函数处理返回值
                channel.writeAndFlush(rpcRequest).addListener(future1 -> {
                    if (future1.isSuccess()) {
                        //todo: 为什么接收回来的信息也是rpcRequest呢？
                        logger.info(String.format("客户端成功发送信息 %s", rpcRequest.toString()));
                    } else {
                        logger.error("接受客户端发送信息失败：%s", future1.cause());
                    }
                });
                channel.closeFuture().sync(); //closeFuture是个Future, 这个future在连接关闭时会完成
                AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse");
                RpcResponse rpcResponse = channel.attr(key).get();
                return rpcResponse.getData();
            }
        } catch (InterruptedException e) {
            logger.error("客户端发送消息时发生错误：", e);
        }
        return null;


    }
}
